package day03

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.4.2	双Value类型交互
 */
object Transformation02Value {

}

/**
 * 2.4.2.1 union()并集
 *
 * 任何相同的元素都会出现多次，必要时使用distinct()去重。
 */
object Transformation_doubleValue_union {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val rdd1: RDD[Int] = sc.makeRDD(1 to 4)

    //4.创建第二个RDD
    val rdd2: RDD[Int] = sc.makeRDD(4 to 8)

    //5. 调用union()取两个RDD的并集
    val unionRDD: RDD[Int] = rdd1.union(rdd2).distinct()

    println("------------------------------------")

    //6.打印输出
    unionRDD.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.2.2 subtract ()差集
 */
object Transformation_doubleValue_subtract {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val rdd1: RDD[Int] = sc.makeRDD(1 to 4)

    //4.创建第二个RDD
    val rdd2: RDD[Int] = sc.makeRDD(4 to 8)

    //5.调用subtract()求差集
    val subtractRdd: RDD[Int] = rdd1.subtract(rdd2)

    //6.打印输出
    subtractRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

/**
 * 2.4.2.3 intersection()交集
 */
object Transformation_doubleValue_intersection {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val rdd1: RDD[Int] = sc.makeRDD(1 to 4)

    //4.创建第二个RDD
    val rdd2: RDD[Int] = sc.makeRDD(4 to 8)

    //5.求两个RDD的交集
    val intersectionRdd: RDD[Int] = rdd1.intersection(rdd2)

    // 打印输出
    intersectionRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()

  }
}

/**
 * 2.4.2.4 zip()拉链
 *
 * 说明：元素个数不同，不能拉链；分区数不同，不能拉链
 */
object Transformation_doubleValue_zip {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val rdd1: RDD[Int] = sc.makeRDD(Array(1, 2, 3), 3)

    //4.创建第二个RDD
    val rdd2: RDD[String] = sc.makeRDD(Array("张三", "李四", "王五"), 3)

    //5.调用zip()
    val rddZip: RDD[(Int, String)] = rdd1.zip(rdd2)

    //6.打印输出
    rddZip.collect().foreach(println)

    // 关闭连接
    sc.stop()
  }
}

